26. 长三角跨境电商销售分析（国际平台样例对照）

import pandas as pd  # 导入表格库
def summarize_platform_discounts(superstore):  # 定义平台折扣分析接口
    required={'Order Date','Sub-Category','Discount','Profit','Quantity','Sales','Product Name'}  # 对齐平台任务字段
    if not required<=set(superstore.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 缺列时停止
    clean=superstore.dropna(subset=list(required)).drop_duplicates().copy()  # 清除关键缺失与完全重复行
    if clean.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空样本时停止
    clean['Order Date']=pd.to_datetime(clean['Order Date'])  # 统一订单日期
    cells=clean.groupby(['Discount','Product Name','Order Date'],as_index=False).agg(quantity=('Quantity','sum'),sales=('Sales','sum'),profit=('Profit','sum'))  # 合并实际出现的折扣产品日单元
    grouped=cells.groupby('Discount').agg(quantity=('quantity','sum'),sales=('sales','sum'),profit=('profit','sum'),observed_cells=('profit','size'),profit_per_product_day=('profit','mean'),sales_per_product_day=('sales','mean'),quantity_per_product_day=('quantity','mean'))  # 在实际观测单元上汇总总额与均值
    if grouped['observed_cells'].eq(0).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 实际观测单元为零时停止
    assert grouped['profit'].sum()==clean['Profit'].sum()==cells['profit'].sum()  # 核对两层聚合利润守恒
    return grouped,{'raw_rows':len(superstore),'clean_rows':len(clean),'profit_total':clean['Profit'].sum()}  # 返回行数与利润核对结果

def select_latest_disclosure(frame,value_fields):  # 统一财报披露版本选择约定
    required={'ts_code','end_date',*value_fields}; versions=[c for c in ['ann_date','f_ann_date'] if c in frame.columns]  # 固定键与可接受版本字段
    if not required<=set(frame.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 缺列时停止
    work=frame.dropna(subset=list(required)).copy(); work['end_date']=pd.to_datetime(work['end_date'],errors='coerce')  # 清理必需字段
    for column in versions: work[column]=pd.to_datetime(work[column],errors='coerce')  # 解析披露日期
    keys=['ts_code','end_date']; work=work.dropna(subset=['end_date']); duplicate_mask=work.duplicated(keys,keep=False); before=int(duplicate_mask.sum())  # 统计选择前重复
    if before and (not versions or work.loc[duplicate_mask,versions].isna().all(axis=1).any()): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 无版本依据时停止
    work['_version']=work[versions].max(axis=1) if versions else pd.NaT  # 合并双版本字段为排序键
    latest_version=work.groupby(keys)['_version'].transform('max') if versions else pd.Series(pd.NaT,index=work.index)  # 为每个公司报告期定位最晚披露时点
    latest_rows=work.loc[~duplicate_mask | work['_version'].eq(latest_version)].copy()  # 仅保留单例或并列最晚版本
    tied_mask=latest_rows.duplicated(keys,keep=False); tied_latest_rows=int(tied_mask.sum())  # 统计并列最晚版本行
    conflicting_latest_keys=int(latest_rows.loc[tied_mask].groupby(keys,dropna=False)[value_fields].nunique(dropna=False).gt(1).any(axis=1).sum()) if tied_latest_rows else 0  # 统计并列且数值冲突的披露键
    if conflicting_latest_keys: raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 并列最新值冲突时请先检查数据后再继续
    selected=latest_rows.drop_duplicates(keys+value_fields).drop_duplicates(keys).drop(columns='_version',errors='ignore')  # 仅合并数值完全一致的并列行
    after=int(selected.duplicated(keys).sum()); audit={'duplicate_rows_before':before,'tied_latest_rows':tied_latest_rows,'conflicting_latest_keys':conflicting_latest_keys,'duplicate_rows_after':after,'version_fields':versions}  # 返回并列与冲突检查
    if after: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 选择后仍重复则停止
    return selected,audit  # 返回唯一版本与检查依据

from pathlib import Path  # 导入路径工具
import pandas as pd  # 导入表格库
import numpy as np  # 导入有限值检查工具
root=Path('/home/ubuntu/r2_data_mount/data/stock')  # 指定股票目录
basic_path,fs_path=root/'stock_basic_data.h5',root/'financial_statement.h5'  # 指定可用HDF资产
if not basic_path.exists() or not fs_path.exists(): raise FileNotFoundError('未找到课程数据文件，请从课程数据下载入口获取并核对文件位置')  # 缺失即停止
basic=pd.read_hdf(basic_path,key='stock_basic_info',columns=['order_book_id','symbol','province']).rename(columns={'order_book_id':'raw_code','symbol':'name','province':'area'})  # 只读取公司筛选字段并标准化
basic['ts_code']=basic['raw_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一证券代码后缀
basic['area']=basic['area'].astype('string').str.replace(r'[省市]$','',regex=True)  # 统一省市简称
required_basic={'ts_code','name','area'}  # 定义公司对象字段
required_financial={'ts_code','end_date','revenue','sell_exp','net_profit'}  # 定义销售费用率与净利率字段
focus=basic.loc[basic['ts_code'].eq('002315.SZ'),['ts_code','name','area']]  # 锁定南京焦点科技
if len(focus)!=1: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 对象不唯一时停止
focus_raw_code=basic.loc[basic['ts_code'].eq('002315.SZ'),'raw_code'].iloc[0]  # 取得HDF原始代码以高效筛选
fs=pd.read_hdf(fs_path,key='financial_data',where='order_book_id == focus_raw_code',columns=['order_book_id','quarter','info_date','operating_revenue','selling_expense','net_profit']).rename(columns={'order_book_id':'ts_code','quarter':'end_date','info_date':'ann_date','operating_revenue':'revenue','selling_expense':'sell_exp'})  # 按目标公司读取营收、销售费用与净利润
fs['ts_code']=fs['ts_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一财报证券代码后缀
fs['end_date']=pd.PeriodIndex(fs['end_date'].str.upper(),freq='Q').end_time.normalize()  # 将季度键转换为报告期末日期
if not required_basic<=set(basic.columns) or not required_financial<=set(fs.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 标准化后仍缺列即停止
fs,version_audit=select_latest_disclosure(fs,['revenue','sell_exp','net_profit'])  # 统一保留最晚披露版本
analysis=fs.merge(focus,on='ts_code',how='inner',validate='many_to_one').sort_values('end_date').copy()  # 连接财报与公司信息并固定时间顺序
analysis=analysis.loc[analysis['end_date'].between('2018-01-01','2025-12-31')].copy()  # 严格锁定2018—2025报告期

if analysis.empty or analysis['end_date'].nunique()<5: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空连接或期间不足时停止
if analysis['revenue'].eq(0).any() or not np.isfinite(analysis[['revenue','sell_exp','net_profit']].to_numpy()).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 比率检查要求
analysis['selling_expense_rate']=analysis['sell_exp']/analysis['revenue']  # 计算销售费用率
analysis['net_profit_margin']=analysis['net_profit']/analysis['revenue']  # 计算净利率并与任务口径一致
if not np.isfinite(analysis[['selling_expense_rate','net_profit_margin']].to_numpy()).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 非有限比率停止
analysis['fiscal_year']=analysis['end_date'].dt.year  # 构造财政年键
analysis['fiscal_quarter']=analysis['end_date'].dt.quarter  # 构造财政季度键
if analysis.duplicated(['fiscal_year','fiscal_quarter']).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 披露选择后期间键仍不唯一时停止
prior_revenue=analysis[['fiscal_year','fiscal_quarter','revenue']].rename(columns={'revenue':'prior_year_revenue'}).copy()  # 建立上年同季收入查找表
prior_revenue['fiscal_year']=prior_revenue['fiscal_year']+1  # 将历史年键前移一年以匹配当期
analysis=analysis.merge(prior_revenue,on=['fiscal_year','fiscal_quarter'],how='left',validate='one_to_one')  # 按年与季度精确连接同比分母
analysis['revenue_yoy']=np.where(analysis['prior_year_revenue'].notna() & analysis['prior_year_revenue'].ne(0),analysis['revenue']/analysis['prior_year_revenue']-1,np.nan)  # 仅在上年同季分母可用时计算同比
print(version_audit,analysis[['ts_code','name','area','end_date','revenue','selling_expense_rate','net_profit_margin','revenue_yoy']])  # 输出数据版本检查与双指标表

ratio_check=analysis[['sell_exp','net_profit']].div(analysis['revenue'],axis=0)  # 独立重算销售费用率与净利率
assert ratio_check['sell_exp'].equals(analysis['selling_expense_rate'])  # 核对销售费用率
assert ratio_check['net_profit'].equals(analysis['net_profit_margin'])  # 核对净利率
sensitivity=pd.DataFrame({'selling_rate_ceiling':[.03,.05,.08],'net_margin_floor':[.05,.03,0]})  # 定义三组双最低要求课堂情景
sensitivity['latest_selling_rate']=analysis['selling_expense_rate'].iloc[-1]  # 固定最新报告期费用率
sensitivity['latest_net_margin']=analysis['net_profit_margin'].iloc[-1]  # 固定最新报告期净利率
sensitivity['status']=sensitivity.apply(lambda row:'REVIEW' if row['latest_selling_rate']>row['selling_rate_ceiling'] or row['latest_net_margin']<row['net_margin_floor'] else 'PASS',axis=1)  # 任一最低要求失败即核对
sensitivity['threshold_status']='课堂预算情景，待真实预算批准'  # 标明阈值尚非生产政策
sensitivity['owner']='跨境电商财务负责人'  # 指定核对责任角色
sensitivity['action']=sensitivity['status'].map({'REVIEW':'暂缓扩大使用并核对费用','PASS':'继续观察'})  # 记录建议
sensitivity['recheck_when']='输入字段改变或连续两期超预算'  # 定义升级需要重新检查的情况
sensitivity['recheck_date']='下一季度财报公布后第5个工作日'  # 事先设定再次检查时间
assert analysis['end_date'].between('2018-01-01','2025-12-31').all()  # 独立核验固定期间
print({'n':len(analysis),'start':analysis['end_date'].min(),'end':analysis['end_date'].max()},sensitivity)  # 输出期间与双最低要求

alternative_raw_code='300792.XSHE'  # 事先设定同属互联网服务业的浙江壹网壹创
alternative_financials=pd.read_hdf(fs_path,key='financial_data',where='order_book_id == alternative_raw_code',columns=['order_book_id','quarter','info_date','operating_revenue','selling_expense','net_profit']).rename(columns={'order_book_id':'ts_code','quarter':'end_date','info_date':'ann_date','operating_revenue':'revenue','selling_expense':'sell_exp'})  # 按相同真实字段结构读取新案例公司财报
alternative_financials['ts_code']=alternative_financials['ts_code'].str.replace('.XSHE','.SZ',regex=False)  # 统一新案例公司证券代码后缀
alternative_financials['end_date']=pd.PeriodIndex(alternative_financials['end_date'].str.upper(),freq='Q').end_time.normalize()  # 将新案例公司季度键转为报告期末
alternative_financials,alternative_version_audit=select_latest_disclosure(alternative_financials,['revenue','sell_exp','net_profit'])  # 按同一规则选择最晚披露版本
alternative_analysis=alternative_financials.loc[alternative_financials['end_date'].between('2018-01-01','2025-12-31')].sort_values('end_date').copy()  # 严格锁定新案例公司2018—2025期间
if alternative_analysis.empty or alternative_analysis['revenue'].eq(0).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 没有有效分母时停止
alternative_analysis['selling_expense_rate']=alternative_analysis['sell_exp']/alternative_analysis['revenue']  # 按相同口径计算新案例销售费用率
alternative_analysis['net_profit_margin']=alternative_analysis['net_profit']/alternative_analysis['revenue']  # 按相同口径计算新案例净利率
alternative_latest=alternative_analysis.iloc[-1]  # 取得新案例公司最新合规报告期
alternative_status='REVIEW' if alternative_latest['selling_expense_rate']>.05 or alternative_latest['net_profit_margin']<.03 else 'PASS'  # 使用同一基准双最低要求形成新案例状态
assert alternative_analysis['end_date'].between('2018-01-01','2025-12-31').all()  # 核验新案例样本仍在固定期间
print({'source_rows':len(analysis),'alternative_rows':len(alternative_analysis),'source_latest':analysis[['selling_expense_rate','net_profit_margin']].iloc[-1].to_dict(),'alternative_latest':alternative_latest[['selling_expense_rate','net_profit_margin']].to_dict(),'alternative_status':alternative_status},alternative_version_audit)  # 输出新案例前后结构、比率与数据版本检查